package com.idanchuang.component.redis.util.pubsub;

import com.idanchuang.component.redis.util.DRedisLock;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author yjy
 * Created at 2022/2/22 1:22 下午
 */
public class LockReleasePubSubs implements MessageListener {

    public static final ExecutorService ASYNC = new ThreadPoolExecutor(1, 1,
            1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1000),
            (runnable) -> new Thread(runnable, "lock-async"));

    public static final String CHANNEL = "LockReleasePubSubs:{" + DRedisLock.HASH_TAG + "}";
    private static RedisConnectionFactory redisConnectionFactory;
    private static MessageListener messageListener;

    private static final AtomicBoolean INIT_COMPLETED = new AtomicBoolean(false);

    public static final int HASH_QUEUE_SIZE = 50;
    private static final HashQueue[] HASH_QUEUES = new HashQueue[HASH_QUEUE_SIZE];

    public LockReleasePubSubs(RedisConnectionFactory redisConnectionFactory) {
        LockReleasePubSubs.redisConnectionFactory = redisConnectionFactory;
        LockReleasePubSubs.messageListener = this;
    }

    static {
        for (int i = 0; i < HASH_QUEUES.length; i++) {
            HASH_QUEUES[i] = new HashQueue();
        }
    }

    /**
     * 初始化事件订阅
     */
    private static void init() {
        if (INIT_COMPLETED.get()) {
            return;
        }
        if (!INIT_COMPLETED.compareAndSet(false, true)) {
            return;
        }
        // 发起订阅
        redisConnectionFactory.getConnection().subscribe(messageListener, CHANNEL.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 订阅锁释放消息
     *
     * @param lockSemaphore 锁释放后的回调逻辑
     */
    public static void register(LockSemaphore lockSemaphore) {
        init();
        lockSemaphore.setAlive(true);
        getContent(lockSemaphore.getLockKey()).add(lockSemaphore);
    }

    /**
     * 取消订阅锁释放消息
     *
     * @param lockSemaphore 回调对象
     */
    public static void unregister(LockSemaphore lockSemaphore) {
        lockSemaphore.setAlive(false);
        ASYNC.execute(() -> getContent(lockSemaphore.getLockKey()).remove(lockSemaphore));
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String key = new String(message.getBody());
        LockSemaphore lockSemaphore = getContent(key).poll(key);
        if (lockSemaphore != null) {
            lockSemaphore.release();
        }
    }

    private static HashQueue getContent(String key) {
        return HASH_QUEUES[Math.abs(key.hashCode()) % HASH_QUEUE_SIZE];
    }

}
